package com.innodealing.offsetskip.handler;

import com.innodealing.offsetskip.config.KafkaOffsetSkipProperties;

/**
 * Kafka Offset Skipper 处理器
 * <p>
 * 负责判断指定的消息是否应该被跳过
 */
public class KafkaOffsetSkipHandler {
    private final KafkaOffsetSkipProperties properties;

    public KafkaOffsetSkipHandler(KafkaOffsetSkipProperties properties) {
        this.properties = properties;
    }

    /**
     * 判断指定的消息是否应该跳过
     *
     * @param groupId       消费者组ID
     * @param topic         主题名称
     * @param partition     分区号
     * @param offset 当前消息的offset
     * @return true表示应该跳过，false表示不应该跳过
     */
    public boolean shouldSkip(String groupId, String topic, int partition, long offset) {
        String key = String.format("%s.%s.%d", groupId, topic, partition);
        Long configured = properties.getConfig().get(key);
        return configured != null && offset < configured;
    }

    /**
     * 判断是否为最后一条消息
     * @param groupId       消费者组ID
     * @param topic         主题名称
     * @param partition     分区号
     * @param offset 当前消息的offset
     * @return true表示为最后一条消息，false表示不是最后一条消息
     */
    public boolean isLastMessageSkipped(String groupId, String topic, int partition, long offset) {
        String key = String.format("%s.%s.%d", groupId, topic, partition);
        final long nextOffset = offset + 1;
        Long configured = properties.getConfig().get(key);
        return configured != null && nextOffset >= configured;
    }
}
